package com.shujia.flink.source;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;

/**
 * @author SHUJIA
 */
public class Demo7TextFormat {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //构建text format source
        FileSource<String> fileSource = FileSource
                //读取数据的格式和路径
                .forRecordStreamFormat(new TextLineInputFormat(), new Path("data/words"))
                //循环监控目录的间隔时间
                .monitorContinuously(Duration.ofSeconds(1))
                .build();

        DataStream<String> lines = env
                .fromSource(fileSource, WatermarkStrategy.noWatermarks(), "fileSource");

        lines.print();

        env.execute();
    }
}
